package main

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"sync"
)

// MyStream 结构体，其中包含数据、位置、通道和互斥锁。
type MyStream struct {
	buffer bytes.Buffer
	ch     chan struct{}
	mu     sync.RWMutex
}

// NewMyStream 初始化一个新的MyStream实例。
func NewMyStream() *MyStream {
	return &MyStream{ch: make(chan struct{})}
}

// Write 将字节数据写入流中，同时更新通道。
func (s *MyStream) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	payload := make([]byte, 4+len(p))
	binary.BigEndian.PutUint32(payload[:4], uint32(len(p)))
	copy(payload[4:], p)

	n, err := s.buffer.Write(payload)
	if err == nil {
		s.ch <- struct{}{}
	}
	return n, err
}

// Read 从流中读取字节数据。
func (s *MyStream) Read(p []byte) (int, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	for {
		if s.buffer.Len() < 4 {
			select {
			case <-s.ch:
				continue
			}
		}

		var payloadSize uint32
		binary.Read(io.LimitReader(&s.buffer, 4), binary.BigEndian, &payloadSize)

		if uint32(s.buffer.Len()) < payloadSize+4 {
			select {
			case <-s.ch:
				continue
			}
		}

		s.buffer.Next(4)
		n, err := s.buffer.Read(p)
		if err != nil {
			return 0, err
		}
		return n, nil
	}
}

// Close 关闭流。
func (s *MyStream) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.buffer.Reset()
	close(s.ch)
	return nil
}

// 测试主函数
func main() {
	s := NewMyStream()

	var wg sync.WaitGroup
	wg.Add(2)

	// 向流中写入数据
	go func() {
		defer wg.Done()
		s.Write([]byte("hello world"))
	}()

	go func() {
		defer wg.Done()
		// 从流中读取数据并输出到标准输出
		buf := make([]byte, 1024)
		n, err := s.Read(buf)
		if err != nil {
			panic(err)
		}
		fmt.Println(string(buf[:n]))
	}()

	wg.Wait()
	// 关闭流
	s.Close()
}

// func main() {
// 	s := NewMyStream()

// 	var wg sync.WaitGroup
// 	wg.Add(2)

// 	// 向流中写入数据
// 	go func() {
// 		defer wg.Done()
// 		s.Write([]byte("hello world"))
// 	}()

// 	go func() {
// 		defer wg.Done()
// 		// 从流中读取数据并输出到标准输出
// 		buf := make([]byte, 1024)
// 		n, err := s.Read(buf)
// 		if err != nil {
// 			panic(err)
// 		}
// 		fmt.Println(string(buf[:n]))
// 	}()

// 	wg.Wait()
// 	// 关闭流
// 	s.Close()
// }

// 重写主函数函数，现在会死锁，用中文注释
//func main() {
//	s := NewMyStream()
//
//	var wg sync.WaitGroup
//	wg.Add(2)
//
//	// 向流中写入数据
//	go func() {
//		defer wg.Done()
//		s.Write([]byte("hello world"))
//	}()
//
//	go func() {
//		defer wg.Done()
//		// 从流中读取数据并输出到标准输出
//		buf := make([]byte, 1024)
//		n, err := s.Read(buf)
//		if err != nil {
//			panic(err)
//		}
//		fmt.Println(string(buf[:n]))
//	}()
//
//	wg.Wait()
//	// 关闭流
//	s.Close()
//}
